/*
 * BufferQueue.h
 *
 *  Created on: Jun 14, 2010
 *      Author: howan
 */

#ifndef BUFFERQUEUE_H_
#define BUFFERQUEUE_H_

//std include
#include <iostream>

//3rd party library include
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/thread/thread.hpp>

//subsystem include
#include "defs.h"
#include "log.h"

template<typename Record>
class BufferQueue
    {
public:
    typedef void (*RxCallback)(void * pUserParm);

public:
    BufferQueue(boost::asio::io_service& io, UInt32 queueSizeIn,
            RxCallback cbIn = NULL, void * pParmIn = NULL,
            UInt32 timeOutValueIn = TIME_OUT_MILLIS);
    ~BufferQueue();

    bool startWork();
    bool insertRecord(const Record &, bool bBlockManner = true);
    bool removeRecord(Record& theRecord);

    bool hasRecord(void) const;
    bool isFull(void);
    UInt32 size(void) const;

private:
    void Rx();
    bool findEmptyPosition(Record** ptheRecord);
    void increaseHead(void);

    static const UInt32 QUEUE_UNUSED_SIZE = 1;
    static const UInt32 TIME_OUT_MILLIS = 20;
    static const UInt32 RETRY_SLEEP_TIME_MILLIS = 20;

    volatile UInt32 queueHead;
    volatile UInt32 queueTail;
    UInt32 queueSize;
    UInt32 queueWrapSize;

    boost::asio::deadline_timer timer_;
    UInt32 timeOutValue;

    Record * pQueueData;

    RxCallback userCb;
    void * pUserParm;
    };

template<typename Record>
inline BufferQueue<Record>::BufferQueue(boost::asio::io_service& io,
        UInt32 queueSizeIn, RxCallback cbIn, void * pParmIn,
        UInt32 timeOutValueIn) :
    timer_(io, boost::posix_time::milliseconds(timeOutValueIn))
    {
    queueHead = 0;
    queueTail = 0;

    userCb = cbIn;
    pUserParm = pParmIn;
    timeOutValue = timeOutValueIn;
    queueSize = queueSizeIn;
    pQueueData = new Record[queueSize];
    queueWrapSize = queueSize - 1;
    }

template<typename Record>
inline BufferQueue<Record>::~BufferQueue<Record>()
    {
    if (pQueueData)
        {
        delete[] pQueueData;
        pQueueData = NULL;
        }
    }

//Get the number of records till now
template<typename Record>
inline UInt32 BufferQueue<Record>::size(void) const
    {
    UInt32 myHead = queueHead;
    UInt32 myTail = queueTail;

    if (myHead >= myTail)
        return myHead - myTail;

    return ((queueWrapSize - myTail) + myHead);
    }

//Find the 1st empty postion in BufferQueue.
//If exists, return the postion and true
//else return false.
template<typename Record>
inline bool BufferQueue<Record>::findEmptyPosition(Record** pRecord)
    {
    UInt32 freeSpace = 0;
    UInt32 myHead = queueHead;
    UInt32 myTail = queueTail;
    bool status = false;

    freeSpace = myTail - myHead;

    if (myHead >= myTail)
        {
        freeSpace = (queueWrapSize - myHead) + myTail;
        }

    if (freeSpace > QUEUE_UNUSED_SIZE)
        {
        *pRecord = &pQueueData[myHead];
        status = true;
        }

    return (status);
    }

template<typename Record>
inline void BufferQueue<Record>::increaseHead(void)
    {
    UInt32 myHead = queueHead + 1;
    UInt32 tmp = 0;

    if (myHead < queueWrapSize)
        tmp = myHead;

    //Keep only one write to queueHead to be atomic operation
    queueHead = tmp;
    }

template<typename Record>
inline bool BufferQueue<Record>::hasRecord(void) const
    {
    if (queueHead != queueTail)
        return (true);
    return (false);
    }

template<typename Record>
inline bool BufferQueue<Record>::isFull(void)
    {
    Record * pTmp = NULL;
    return findEmptyPosition(&pTmp) ? false : true;
    }

//Remove current record(tail point to) to input paramerter
template<typename Record>
inline bool BufferQueue<Record>::removeRecord(Record& theRecord)
    {
    UInt32 myHead = queueHead;
    UInt32 myTail = queueTail;
    bool bHasRecord = false;
    UInt32 tmp = 0;

    if (myHead != myTail)
        {
        // Get the record
        bHasRecord = true;
        theRecord = pQueueData[myTail];

        ++myTail;
        if (myTail < queueWrapSize)
            tmp = myTail;

        // Only one write to queueTail ever
        queueTail = tmp;
        }

    return (bHasRecord);
    }

template<typename Record>
inline bool BufferQueue<Record>::startWork()
    {
    timer_.async_wait(boost::bind(&BufferQueue<Record>::Rx, this));

    return true;
    }

template<typename Record>
inline bool BufferQueue<Record>::insertRecord(const Record & recordBlock,
        bool bBlockManner)
    {
    Record * pSlot = NULL;

    //...If bBlockManner == true, we don't return until we get a slot
    while (findEmptyPosition(&pSlot) == false)
        {
        if (bBlockManner == false)
            {
            //...If we don't block, we immediately return upon failure
            return false;
            }

        //...Sleep for a short period of time and try again, so we don't take up all the cpu time
        boost::this_thread::sleep(boost::posix_time::milliseconds(
                RETRY_SLEEP_TIME_MILLIS));
        }

    *pSlot = recordBlock;
    increaseHead();
    return true;
    }

template<typename Record>
inline void BufferQueue<Record>::Rx()
    {
    LOGCON.infoStream() << __PRETTY_FUNCTION__ << " - Timeout again\n";

    if (hasRecord() && userCb)
        (*userCb)(pUserParm);

    //...No more data, start the polling timer
    timer_.expires_at(timer_.expires_at() + boost::posix_time::milliseconds(
            timeOutValue));
    timer_.async_wait(boost::bind(boost::bind(&BufferQueue<Record>::Rx, this)));
    }

#endif /* BUFFERQUEUE_H_ */
